package com.udf.hive.udtf;


import org.apache.hadoop.hive.ql.exec.Description;
import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
import org.apache.hadoop.hive.serde2.objectinspector.MapObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;

import java.util.ArrayList;
import java.util.Map;

/**
 * facebookudf
 * https://github.com/brndnmtthws/facebook-hive-udfs/tree/master/src/main/java/com/facebook/hive/udf
 * 
 * map数据转换成多行key、value形式
 * uid attr 
 * 1 {"name":"zhangsan","sex":"female","age":18}
 * 2 {"name":"lisi","sex":"male"} 
 * 转换后 
 * uid col val 
 * 1 name zhangsan 
 * 1 sex female 
 * 1 age 18 
 * 2 name lisi 
 * 2 sex male
 * 
 * 添加临时函数
 * create temporary function explode_map as 'com.smzdm.functions.udtf.UDTFExplodeMap';
 * 
 * udtf用法【https://www.cnblogs.com/ggjucheng/archive/2013/02/01/2888819.html】
	1：直接使用
	select explode_map(result) as (col1,col2) from cbd_cds.cds_clt_log_combine_result limit 10;
	
	2：和lateral view一起使用
	select url, mytable.col1, mytable.col2 from cbd_cds.cds_clt_log_combine_result lateral view explode_map(result) mytable as col1, col2 limit 10;

 * 
 * 
 * GenericUDTFExplodeMap. map数据打散。
 * 
 * @author admin
 * @version 2017-11-20
 */
@Description(name = "explode_map", value = "_FUNC_(a) - separates the elements of map a into multiple rows ",
			 extended = "Example:\n"
					  + "  > 1. SELECT explode_map(field) as (col1,col2) FROM tmp;\n"
					  + "  > 2. SELECT url, mytable.col1, mytable.col2 FROM tmp LATERAL VIEW explode_map(map字段) mytable as col1, col2 limit 10;\n")
public class UDTFExplodeMap extends GenericUDTF {

	private MapObjectInspector mapOI = null;

	@Override
	public void close() throws HiveException {
	}

	@Override
	public StructObjectInspector initialize(ObjectInspector[] args)
			throws UDFArgumentException {
		if (args.length != 1) {
			throw new UDFArgumentException(
					"explode_map() takes only one argument");
		}

		if (args[0].getCategory() != ObjectInspector.Category.MAP) {
			throw new UDFArgumentException(
					"explode_map() takes a map as a parameter");
		}
		mapOI = (MapObjectInspector) args[0];

		ArrayList<String> fieldNames = new ArrayList<String>();
		ArrayList<ObjectInspector> fieldOIs = new ArrayList<ObjectInspector>();
		fieldNames.add("col1");
		fieldNames.add("col2");
		fieldOIs.add(mapOI.getMapKeyObjectInspector());
		fieldOIs.add(mapOI.getMapValueObjectInspector());
		return ObjectInspectorFactory.getStandardStructObjectInspector(
				fieldNames, fieldOIs);
	}

	private final Object[] forwardObj = new Object[2];

	@Override
	public void process(Object[] o) throws HiveException {
		Map<?, ?> map = mapOI.getMap(o[0]);
		if (map == null) {
			return;
		}
		for (Map.Entry entry : map.entrySet()) {
			forwardObj[0] = entry.getKey();
			forwardObj[1] = entry.getValue();
			forward(forwardObj);
		}
	}

	@Override
	public String toString() {
		return "explode_map";
	}
}
